package org.apache.flink.kubernetes.kubeclient.decorators;

import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesMasterComponentParameters;
import org.apache.flink.kubernetes.utils.Constants;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

import static org.apache.flink.util.Preconditions.checkNotNull;

/** Creates an external Service to expose the rest port of the master. */
public class ExternalServiceDecorator extends AbstractKubernetesStepDecorator {

    private final KubernetesMasterComponentParameters kubernetesJobManagerParameters;

    public ExternalServiceDecorator(KubernetesMasterComponentParameters kubernetesJobManagerParameters) {
        this.kubernetesJobManagerParameters = checkNotNull(kubernetesJobManagerParameters);
    }

    @Override
    public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {
        final Service service =
                kubernetesJobManagerParameters
                        .getRestServiceExposedType()
                        .serviceType()
                        .buildUpExternalRestService(kubernetesJobManagerParameters);

        return Collections.singletonList(service);
    }

    /** Generate name of the external rest Service. */
    public static String getExternalServiceName(String clusterId) {
        return clusterId + Constants.FLINK_REST_SERVICE_SUFFIX;
    }

    /**
     * Generate namespaced name of the external rest Service by cluster Id, This is used by other
     * project, so do not delete it.
     */
    public static String getNamespacedExternalServiceName(String clusterId, String namespace) {
        return getExternalServiceName(clusterId) + "." + namespace;
    }
}
